package atlub.tweet_crawler;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.BlockingQueue;

import org.mybeans.dao.DAOException;

import atlub.tweet_crawler.datamodel.Tweet;
import atlub.tweet_crawler.datamodel.Twitter_user;
import atlub.tweet_crawler.datamodel.Url;

import twitter4j.Paging;
import twitter4j.Status;
import twitter4j.Twitter;
import twitter4j.TwitterException;
import twitter4j.URLEntity;
import twitter4j.User;

import zunge.tool.db.sql.BeanDAO;
import zunge.tool.db.sql.DbConn;

public class CrawlerThread extends Thread{
    static final String host = "islpc2.is.cs.cmu.edu";
    static final String db = "social_media";
    static final String user = "smuser";
    static final String pswd = "socialmedia";
    
    static final String tweetTable = "btweet";
    static final String userTable = "buser";
    static final String urlTable = "burl";
    
    Twitter twitter = null;
    DbConn conn = null;
    BeanDAO<Tweet> tweetDao = null;
    BeanDAO<Twitter_user> userDao = null;
    BeanDAO<Url> urlDao = null;
    
    CrawlerThreadController controller;
    public BlockingQueue<Long> userPool = null;
    boolean isFinished = false;
    
    public CrawlerThread(CrawlerThreadController controller, BlockingQueue<Long> userPool, Twitter twitter){
	this.controller = controller;
	this.userPool = userPool;
	this.twitter = twitter;
	try{
	    conn = new DbConn(host,db,user,pswd);
	    this.tweetDao = new BeanDAO<Tweet>(Tweet.class, conn.db_url, user, pswd, tweetTable, "tweet_id");
	    this.userDao = new BeanDAO<Twitter_user>(Twitter_user.class, conn.db_url, user,pswd, userTable, "user_id");
	    this.urlDao = new BeanDAO<Url>(Url.class, conn.db_url, user, pswd, urlTable, "tweet_id");
	    
	}catch(SQLException e){
	    e.printStackTrace();
	} catch (InstantiationException e) {
	    e.printStackTrace();
	} catch (IllegalAccessException e) {
	    e.printStackTrace();
	} catch (ClassNotFoundException e) {
	    e.printStackTrace();
	} catch (DAOException e) {
	    e.printStackTrace();
	}
    }
    
    @Override
    public void run() {
	try{
	    
	    while(!isFinished && userPool.size()>0){
		Long userId = userPool.poll();
		if(userId==null){break;}
		processUser(userId);
	    }
	    System.out.println("Thread finished");
	    
	    
	    conn.close();
	}catch(SQLException e){e.printStackTrace();}
    }

    public void finishThread(){
	isFinished = true;
    }
    
    void processUser(long userId){
	//System.out.println("Processing User "+userId);
	crawlUser(userId);
	crawlTweets(userId);
    }
    
    /* Crawling */
    
    public static List<Long> getAllUserIds() throws Exception{
	DbConn conn = new DbConn(host,db,user,pswd);
	
	HashSet<Long> userIds = new HashSet<Long>();
	
	String query = "SELECT DISTINCT user_id1 FROM mutual_follow";
	ResultSet rs = conn.executeQuery(query);
	while(rs.next()){
	    userIds.add(rs.getLong("user_id1"));
	}
	rs.close();
	
	query = "SELECT DISTINCT user_id FROM "+tweetTable;
	rs = conn.executeQuery(query);
	while(rs.next()){
	    userIds.remove(rs.getLong("user_id"));
	}
	rs.close();
	
	conn.close();
	List<Long> ret = new ArrayList<Long>(userIds);
	return ret;
    }
    
    
    
    /**
     * Crawl user information
     * @param userId
     */
    void crawlUser(long userId){
	boolean retried = false;
	boolean error = false;
	User user  = null;
	while(!isFinished){
	    try{
		user = twitter.showUser(userId);
		if(retried){wakeThread();}
		break;
	    }catch(TwitterException te){
		if(te.exceededRateLimitation()){
		    retried = true;
		    sleepThread();
		}else{
		    error = true;
		    System.err.println(te.getStatusCode()+"\t"+te.getErrorMessage());
		    break;
		}
	    }
	}
	if(error || user==null){return;}
	Twitter_user tu = translateUser(user);
	try{
	    userDao.create(tu);
	}catch(DAOException e){System.err.println(e.getMessage());}
    }
    
    /**
     * Crawl the latest 200 tweets
     * @param userId
     */
    void crawlTweets(long userId){
	
	Paging paging = new Paging(1, 200);
	List<Status> statuses = null;
	boolean retried = false;
	boolean error = false;
	while(!isFinished){
	    try{
		statuses = twitter.getUserTimeline(userId, paging);
		if(retried){wakeThread();}
		break;
	    }catch(TwitterException te){
		if(te.exceededRateLimitation()){
		    retried = true;
		    sleepThread();
		}else{
		    error = true;
		    System.err.println(te.getStatusCode()+"\t"+te.getErrorMessage());
		    break;
		}
	    }
	}
	if(error){return;}
	for(Status status:statuses){
	    Tweet tweet = translateStatus(status);
	    try{
		tweetDao.create(tweet);
		//TODO:fetch url
	    }catch(Exception e){System.err.println(e.getMessage());}
	}
    }
    
    /* Bean translation */
    
    public static Twitter_user translateUser(User user){
	Twitter_user tu = new Twitter_user();
	tu.setUser_id(user.getId());
	tu.setScreen_name(user.getName());
	tu.setGeo_enabled(user.isGeoEnabled()?1:0);
	tu.setUtc_offset(user.getUtcOffset());
	tu.setFollowers_count(user.getFollowersCount());
	tu.setFriends_count(user.getFriendsCount());
	tu.setListed_count(user.getListedCount());
	tu.setStatuses_count(user.getStatusesCount());
	tu.setFavourites_count(user.getFavouritesCount());
	return tu;
    }
    
    public static Tweet translateStatus(Status status){
	Tweet tweet = new Tweet();
	tweet.setTweet_id(status.getId());
	tweet.setUser_id(status.getUser().getId());
	tweet.setText(status.getText());
	tweet.setCreated_at(status.getCreatedAt());
	tweet.setSource(status.getSource());
	tweet.setIn_reply_to_status_id(status.getInReplyToStatusId());
	tweet.setIn_reply_to_user_id(status.getInReplyToUserId());
	if(status.getGeoLocation()!=null){
	    tweet.setGeo(status.getGeoLocation().getLatitude()+","+status.getGeoLocation().getLongitude());//lat,lon
	}
	if(status.getPlace()!=null){
	    tweet.setPlace(status.getPlace().getFullName());
	}
	tweet.setRetweet_count((int)status.getRetweetCount());
	return tweet;
    }
    public static Url[] translateUrl(Status status){
	URLEntity[] entities = status.getURLEntities();
	if(entities == null || entities.length == 0){return null;}
	
	Url[] urls = new Url[entities.length];
	for(int i = 0;i < entities.length;++i){
	    Url url = new Url();
	    url.setTweet_id(status.getId());
	    url.setUser_id(status.getUser().getId());
	    url.setUrl(entities[i].getDisplayURL());
	    urls[i] = url;
	}
	return urls;
    }
    
    /* Thread sleep and wakeup if exceed limit */
    
    int[] sleepSeconds = {60, 120, 240, 480, 60*30}; //if exceeded, check every 15 minutes
    int sleepPtr = 0;
    void sleepThread(){
	try{
	    System.err.println("Sleep for "+sleepSeconds[sleepPtr]+" seconds.");
	    Thread.sleep(sleepSeconds[sleepPtr]*1000);
	    sleepPtr =(  (sleepPtr == sleepSeconds.length-1)?sleepPtr:(sleepPtr+1)   )%sleepSeconds.length;
	}catch(InterruptedException e){e.printStackTrace();}
    }
    void wakeThread(){
	sleepPtr = 0;
    }
}
